package com.sxzjrj.mblybc

import com.sxzjrj.config.MBYLBCConfig
import com.sxzjrj.constant.Constant
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
  * Created by ljj in 2019/3/25
  *
  */
object MBYLBCCountSparkSql {

  def main(args: Array[String]): Unit = {

    val session = SparkSession
      .builder()
      .appName("慢病医疗报表")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.parquet.compression.codec", "snappy")
      .getOrCreate()

    import session.implicits._
    val data = session.read.textFile(MBYLBCConfig.srcPath)
      .map(_.split("\t", -1))
      .filter(arr => arr(1).length > 0 && arr(2).length > 0 && arr(4).length > 0 && arr.length >= Constant.DATALENGTH.toString.toInt)
      .map(arr => {
        val F_GRBM = arr(1) //个人编码
        val F_JZLX = arr(2) //就诊类型
        //val F_YLJG = arr(4) //医疗机构
        //val F_ID = if (arr(29).length > 0) arr(29) else "身份证信息缺失" //身份证号
        var F_SJ_JE = 0.0
        var F_HS_JE = 0.0
        var F_MB_YYF = 0.0
        var F_JZCS = 0
        try {
          F_SJ_JE = if (arr(41).length > 0) arr(41).toDouble else 0.0 //实际费用
          F_HS_JE = if (arr(42).length > 0) arr(42).toDouble else F_SJ_JE //核算金额
          F_MB_YYF = if (arr(43).length > 0) arr(43).toDouble else F_HS_JE //慢病总费用
          F_JZCS = if (F_SJ_JE == 0.0 && F_HS_JE == 0.0 && F_MB_YYF == 0.0) 0 else 1
        } catch {
          case e: Exception => e.printStackTrace()
        }

        (F_GRBM, F_JZLX, F_JZCS, F_SJ_JE, F_HS_JE, F_MB_YYF)
      }).toDF("F_GRBM", "F_JZLX", "F_JZCS", "F_SJ_JE", "F_HS_JE", "F_MB_YYF")

    data.createTempView("t_mzlybc")

    session.sql(
      """
        |select F_GRBM, F_JZLX, sum(F_JZCS) as F_JZCS, sum(F_SJ_JE) as F_SJ_JE,
        |sum(F_HS_JE) as F_HS_JE, sum(F_MB_YYF) as F_MB_YYF from t_mzlybc
        |group by F_GRBM, F_JZLX
      """.stripMargin).repartition(2)
      .write.mode(SaveMode.Append)
      .parquet(MBYLBCConfig.descPath)
      //.jdbc(MZLYBCConfig.url,"t_mzlybc",MZLYBCConfig.properties)
    //      .foreachPartition(iter => {
    //        iter.foreach(println(_))
    //      })


    session.stop()

  }

}
